﻿using Confluent.Kafka;
using Leo.Common.MQ;
using System.Text;
using Leo.MSATracert;
using static Confluent.Kafka.ConfigPropertyNames;

namespace Leo.Common.Kafka
{
    public class KafkaProducer : IKafkaProducer, IMQProducer
    {

        internal const string TYPE_FIELD_NAME = "_t_";


        internal const string TraceIdName = "traceId";


        internal const string CorrelationIdName = "CorrelationId";


        internal const string MessageIdName = "MessageId";


        private IProducer<Null, string> _producer;


        private readonly KafkaSetting setting;

        public MessageQueueType QueueType
        {
            get
            {
                return MessageQueueType.Kafka;
            }
        }


        public KafkaProducer(KafkaSetting setting)
        {
            this.setting = setting;
        }


        private void Init()
        {
            if (this._producer == null)
            {
                ProducerConfig producerConfig = new ProducerConfig
                {
                    BootstrapServers = this.setting.Broker
                };
                this._producer = new ProducerBuilder<Null, string>(producerConfig).Build();
            }
        }


        public Task<bool> ProduceAsync(IMessageModel message)
        {
            return this.SingleProduceAsync(message);
        }


        public Task<bool> ProduceAsync<T>(T data, string destination)
        {
            this.Init();
            MessageModel<T> message = new MessageModel<T>(destination, data, null);
            return this.ProduceAsync(message);
        }


        Task<bool> IMQProducer.ProduceAsync<T>(T data, string destination, MessageProperty property)
        {
            return this.ProduceAsync<T>(data, destination);
        }


        private Task<bool> SingleProduceAsync(IMessageModel message)
        {
             
            //_producer.ProduceAsync(topicName, new Message<string, string>
            //{
            //    Key = Guid.NewGuid().ToString(),
            //    Value = JsonConvert.SerializeObject(message)
            //}); ;

        }


        private void SetMessageHeader(Message<Null, string> payload, IMessageModel message)
        {
            try
            {
                string messageType = message.GetMessageType();
                MessageProperty properties = message.GetProperties();
                string text = (properties != null) ? properties.CorrelationId : null;
                string text2 = (properties != null) ? properties.MessageId : null;
                if (payload.Headers == null)
                {
                    payload.Headers = new Headers();
                }
                if (!string.IsNullOrEmpty(messageType))
                {
                    payload.Headers.Add("_t_", Encoding.UTF8.GetBytes(messageType));
                }
                if (!string.IsNullOrEmpty(text))
                {
                    payload.Headers.Add("CorrelationId", Encoding.UTF8.GetBytes(text));
                }
                if (!string.IsNullOrEmpty(text2))
                {
                    payload.Headers.Add("MessageId", Encoding.UTF8.GetBytes(text2));
                }
                payload.Headers.Add("traceId", Encoding.UTF8.GetBytes(ServiceInvocationLinkTrace.GetPrincipal(null, null).RequestId));
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                Console.WriteLine(ex.StackTrace);
            }
        }

    }
}
